package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/louloulin/dataflare/pkg/cluster/clusterapi"
	_ "github.com/louloulin/dataflare/pkg/cluster/memory"
	"github.com/louloulin/dataflare/pkg/workflow"
	"github.com/louloulin/dataflare/pkg/workflow/distributed"
)

func main() {
	fmt.Println("=== DataFlare Distributed Runtime Verification ===")

	// Create verification directory
	err := os.MkdirAll("verification/data", 0755)
	if err != nil {
		log.Fatalf("Failed to create verification directory: %v", err)
	}

	// Create test workflow file
	workflowPath := "verification/data/distributed-workflow.yaml"
	workflowContent := `
version: "1.0"
name: "distributed-test-workflow"
description: "Distributed test workflow for verification"

sources:
  memory-source:
    type: memory
    config:
      data:
        - {"id": 1, "value": 10}
        - {"id": 2, "value": 20}
        - {"id": 3, "value": 30}

processors:
  transform-processor:
    type: transform
    config:
      function: |
        function process(data) {
          const obj = JSON.parse(data);
          obj.value = obj.value * 2;
          obj.processed = true;
          return JSON.stringify(obj);
        }

sinks:
  console-sink:
    type: console
    config:
      format: json

edges:
  - from: "memory-source"
    to: "transform-processor"
  - from: "transform-processor"
    to: "console-sink"
`
	err = os.WriteFile(workflowPath, []byte(workflowContent), 0644)
	if err != nil {
		log.Fatalf("Failed to create test workflow file: %v", err)
	}

	// Create cluster configuration
	fmt.Println("Creating cluster configuration...")
	config := clusterapi.DefaultConfig()
	config.NodeID = "node1"
	config.BindAddress = "127.0.0.1:7946"
	config.AdvertiseAddress = "127.0.0.1:7946"

	// Create cluster manager
	fmt.Println("Creating cluster manager...")
	manager, err := clusterapi.NewManager("memory", config)
	if err != nil {
		log.Fatalf("Failed to create cluster manager: %v", err)
	}

	// Start cluster manager
	ctx := context.Background()
	fmt.Println("Starting cluster manager...")
	err = manager.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start cluster manager: %v", err)
	}
	defer manager.Stop(ctx)

	// Create distributed engine
	fmt.Println("Creating distributed engine...")
	engine, err := distributed.NewEngine(manager)
	if err != nil {
		log.Fatalf("Failed to create distributed engine: %v", err)
	}
	defer engine.Shutdown()

	// Parse workflow
	fmt.Println("Parsing workflow...")
	definition, err := workflow.ParseFromFile(workflowPath)
	if err != nil {
		log.Fatalf("Failed to parse workflow: %v", err)
	}

	// Deploy workflow
	fmt.Println("Deploying workflow...")
	workflowID, err := engine.DeployWorkflow(ctx, definition)
	if err != nil {
		log.Fatalf("Failed to deploy workflow: %v", err)
	}
	fmt.Printf("Workflow deployed with ID: %s\n", workflowID)

	// Start workflow
	fmt.Println("Starting workflow...")
	runID, err := engine.StartWorkflow(ctx, workflowID, nil)
	if err != nil {
		log.Fatalf("Failed to start workflow: %v", err)
	}
	fmt.Printf("Workflow started with run ID: %s\n", runID)

	// Wait for workflow to complete
	fmt.Println("Waiting for workflow to complete...")
	maxWaitTime := 30 * time.Second
	startTime := time.Now()

	for {
		if time.Since(startTime) > maxWaitTime {
			fmt.Println("Workflow execution timed out")
			break
		}

		status, err := engine.GetWorkflowStatus(ctx, runID)
		if err != nil {
			log.Printf("Failed to get workflow status: %v", err)
			time.Sleep(1 * time.Second)
			continue
		}

		fmt.Printf("Workflow status: %s (Progress: %d%%)\n", status.Status, status.Progress)

		if status.Status == "completed" || status.Status == "failed" {
			break
		}

		time.Sleep(1 * time.Second)
	}

	// Check cluster status
	fmt.Println("\nChecking cluster status...")
	nodes, err := manager.GetNodes(ctx)
	if err != nil {
		log.Fatalf("Failed to get cluster nodes: %v", err)
	}

	fmt.Printf("Cluster nodes: %d\n", len(nodes))
	for i, node := range nodes {
		fmt.Printf("Node %d: ID=%s, Address=%s, Status=%s\n", i+1, node.ID, node.Address, node.Status)
	}

	// Check if this node is leader
	isLeader, err := manager.IsLeader(ctx)
	if err != nil {
		log.Fatalf("Failed to check if node is leader: %v", err)
	}

	fmt.Printf("Is leader: %v\n", isLeader)

	fmt.Println("\n=== Distributed Runtime Verification Complete ===")
}
